package cli

import (
	"errors"
	"gitee.com/tjccc/gomq/lib/logger"
	"gitee.com/tjccc/gomq/mq/protocol"
	parser "gitee.com/tjccc/gomq/parser/custom"
	"math/rand"
	"net"
	"time"
)

type ListenFunc func(msg *protocol.CustomMsg) error

type Consumer struct {
	id      int32
	address string
	conn    net.Conn
	group   string
	topics  []string
	ackFlag bool
	// pull func
	listenFunc ListenFunc
}

func (c *Consumer) Start() error {
	conn, err := net.Dial("tcp", c.address)
	if err != nil {
		logger.Errorf("connect fail causes %s", err)
		return err
	}
	c.conn = conn
	msgCh := parser.ParseStream(conn)
	go func() {
		for msg := range msgCh {
			if msg.Err != nil {
				logger.Errorf("msg fail causes %s", msg.Err)
				continue
			}
			data := msg.Data
			customMsg := data.(*protocol.CustomMsg)
			err := c.listenFunc(customMsg)
			if err != nil {
				logger.Errorf("consume msg fail: %s", err)
				continue
			}
			if c.ackFlag {
				_ = c.ack(customMsg)
			}
		}
	}()

	return nil
}

func (c *Consumer) ack(msg *protocol.CustomMsg) error {
	logger.Info("ack message")
	return nil
}

func tickPull(conn net.Conn) {
	for _ = range time.Tick(300 * time.Millisecond) {
		go Pull(conn)
	}
}

func Pull(conn net.Conn) {
	// 应该是每300ms 进行一个pull请求
	msg := protocol.CustomMsg{
		Code: protocol.PULL,
	}

	_, err := conn.Write(msg.ToBytes())
	if err != nil {
		logger.Errorf("pull data fail: %s", err)
	}
	logger.Info("pull a message")
}

type ConsumerBuilder struct {
	id         int32
	address    string
	conn       net.Conn
	group      string
	topics     []string
	ackFlag    bool
	listenFunc ListenFunc
}

func NewConsumerBuilder() *ConsumerBuilder {
	return &ConsumerBuilder{
		group:   "",
		topics:  []string{""},
		ackFlag: false,
	}
}

func (cb *ConsumerBuilder) Group(group string) *ConsumerBuilder {
	cb.group = group
	return cb
}

func (cb *ConsumerBuilder) Address(address string) *ConsumerBuilder {
	cb.address = address
	return cb
}

func (cb *ConsumerBuilder) Topics(topics ...string) *ConsumerBuilder {
	cb.topics = append(cb.topics, topics...)
	return cb
}

func (cb *ConsumerBuilder) ListenFunc(listener ListenFunc) *ConsumerBuilder {
	cb.listenFunc = listener
	return cb
}

func (cb *ConsumerBuilder) Build() (*Producer, error) {
	if cb.address == "" {
		return nil, errors.New("address is empty")
	}

	p := &Producer{
		address: cb.address,
		ackFlag: cb.ackFlag,
	}
	r := rand.New(rand.NewSource(time.Now().UnixNano()))
	p.id = r.Int31()
	return p, nil
}
